//package com.example.utils;
//
//import org.apache.pulsar.client.api.*;
//import org.springframework.beans.factory.annotation.Qualifier;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.Resource;
//import java.util.concurrent.TimeUnit;
//
///**
// * @author Lucas
// * date 2024/3/6 10:55
// * description pulsar 公共方法
// */
//@Component
//public class PulsarCommon {
//    @Qualifier("PulsarClient")
//    @Resource
//    private PulsarClient client;
//
//    @Qualifier("PulsarTransactionClient")
//    @Resource
//    private PulsarClient transactionClient;
//
//    /**
//     * 创建一个生产者
//     *
//     * @param topic  topic name
//     * @param schema schema方式
//     * @param <T>    泛型
//     * @return Producer生产者
//     */
//    public <T> Producer<T> createProducer(String topic, Schema<T> schema) {
//
//        try {
//            return client.newProducer(schema)
//                    .topic(topic)
//                    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
//                    .sendTimeout(10, TimeUnit.SECONDS)
//                    .blockIfQueueFull(true)
//                    .create();
//        } catch (PulsarClientException e) {
//            throw new RuntimeException("初始化Pulsar Producer失败: " + topic, e);
//        }
//    }
//
//    /**
//     * @param topic           topic name
//     * @param subscription    sub name
//     * @param messageListener MessageListener的自定义实现类
//     * @param schema          schema消费方式
//     * @param <T>             泛型
//     * @return Consumer消费者
//     */
//    public <T> Consumer<T> createConsumer(String topic, String subscription,
//                                          MessageListener<T> messageListener, Schema<T> schema) {
//        try {
//            return client.newConsumer(schema)
//                    .topic(topic)
//                    .subscriptionName(subscription)
//                    .ackTimeout(10, TimeUnit.SECONDS)
//                    .subscriptionType(SubscriptionType.Shared)
//                    .messageListener(messageListener)
//                    .subscribe();
//        } catch (PulsarClientException e) {
//            throw new RuntimeException("初始化Pulsar Consumer失败: " + topic, e);
//        }
//    }
//
//    /**
//     * @param topic           topic name
//     * @param subscription    sub name
//     * @param messageListener MessageListener的自定义实现类
//     * @param schema          schema消费方式
//     * @param <T>             泛型
//     * @return Consumer消费者
//     */
//    public <T> Consumer<T> createConsumerDelay(String topic, String subscription,
//                                               MessageListener<T> messageListener, Schema<T> schema,
//                                               Integer ackRedeliveryTime, TimeUnit ackRedeliveryTimeUnit,
//                                               Integer negativeAckRedeliveryTime, TimeUnit negativeAckRedeliveryTimeUnit) {
//        try {
//            return client.newConsumer(schema)
//                    .topic(topic)
//                    .subscriptionName(subscription)
//                    .ackTimeout(ackRedeliveryTime, ackRedeliveryTimeUnit)
//                    .negativeAckRedeliveryDelay(negativeAckRedeliveryTime, negativeAckRedeliveryTimeUnit)
//                    .subscriptionType(SubscriptionType.Shared)
//                    .messageListener(messageListener)
//                    .subscribe();
//        } catch (PulsarClientException e) {
//            throw new RuntimeException("初始化Pulsar Consumer失败: " + topic, e);
//        }
//    }
//}
